/*********************************************************************************
 *      Copyright:  (C) 2021 Simply myself<1239726156@qq.com>
 *                  All rights reserved.
 *
 *       Filename:  my_mult_thread.c
 *    Description:  This file 
 *                 
 *        Version:  1.0.0(2021年08月09日)
 *         Author:  Simply myself <1239726156@qq.com>
 *      ChangeLog:  1, Release initial version on "2021年08月09日 16时18分55秒"
 *                 
 ********************************************************************************/
 
#include "my_mult_thread.h"

/* 创建一个相分离的线程 */
int thread_start(pthread_t *tid, THREAD_FUNC *thread_workfunc, void *thread_arg)
{
	int             rv = 0;
	pthread_attr_t  thread_attr;
	
	if ( !tid || !thread_workfunc || !thread_arg )
	{
		printf ("[ERROR]: Invalid parameter in %s\n\n", __FUNCTION__);
		return -1;
	}
	
	//初始化线程属性
	if (pthread_attr_init(&thread_attr))
	{
		printf("pthread_attr_init() failure: %s\n", strerror(errno));
		rv = -2;	goto cleanup;
	}
	
	//设置栈大小
	if (pthread_attr_setstacksize(&thread_attr, 120*1024))
	{
		printf("pthread_attr_setstacksize() failure: %s\n", strerror(errno));
		rv = -3;	goto cleanup;
	}
	
	//设置为相分离，主线程就不需要在子线程退出的时候与其会合
	if (pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED))
	{
		printf("pthread_attr_setdetachstate() failure: %s\n", strerror(errno));
		rv = -4;	goto cleanup;
	}
	
	//创建一个子线程
	if (pthread_create(tid, &thread_attr, thread_workfunc, thread_arg))
	{
		printf("pthread_create failure: %s\n", strerror(errno));
		rv = -5;	goto cleanup;
	}

cleanup:
	//线程属性在使用完之后，调用pthread_attr_destroy把它摧毁释放
	pthread_attr_destroy(&thread_attr);
	return rv;

}

/* 订阅子线程函数 */
void *sub_thread_func(void *thread_arg)
{
	int                  rv = 0;
	struct mosquitto    *mosq = NULL;
	mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
	/* 检查 thread_arg 参数是否为空 */
	if ( !thread_arg )
	{
		printf ("+++[THREAD_SUB] [ERROR]: Invalid parameter in %s\n\n", __FUNCTION__);
		return NULL;
	}
	
	/* 把指针指向thread_arg地址 */
	mqtt = (mqtt_t *)thread_arg;
	zc = mqtt->zlog.zc;
	
	/* 新建mosquitto客户端，不保留消息，传入mqtt指针作为参数 */
	mosq = mosquitto_new(mqtt->mosq_ctx.clientid, true, (void *)mqtt);
    if (mosq == NULL)
    {
        printf ("+++[THREAD_SUB] [ERROR]: mosquitto_new failure\n");
		zlog_error(zc, "+++[THREAD_SUB] mosquitto_new failure\n");
        rv = -3;
        goto cleanup;
    }
    printf ("+++[THREAD_SUB] [SUCCESS]: mosquitto_new success\n\n");
	zlog_notice(zc, "+++[THREAD_SUB] mosquitto_new success");
    
    /* 设置回调函数（连接、断开、订阅、消息回调） */
	zlog_notice(zc, "+++[THREAD_SUB] set mosquitto callback\n");
    mosquitto_connect_callback_set(mosq, sub_on_connect);
    mosquitto_disconnect_callback_set(mosq, sub_on_disconnect);
	mosquitto_subscribe_callback_set(mosq, sub_on_subscribe);
    mosquitto_message_callback_set(mosq, sub_on_message);
	

    /* 设置username、passwd */
    printf ("+++[THREAD_SUB] username: %s\n", mqtt->usr.name);
    printf ("+++[THREAD_SUB] passwd: %s\n", mqtt->usr.passwd);
	zlog_info(zc, "+++[THREAD_SUB] username: %s", mqtt->usr.name);
	zlog_info(zc, "+++[THREAD_SUB] passwd: %s", mqtt->usr.passwd);
    rv = mosquitto_username_pw_set(mosq, mqtt->usr.name, mqtt->usr.passwd);
    if (rv != MOSQ_ERR_SUCCESS)
    {
        printf("+++[THREAD_SUB] [ERROR]: mosquitto_username_pw_set failure\n\n");
		zlog_error(zc, "+++[THREAD_SUB] mosquitto_username_pw_set failure\n");
        rv = -4;
        goto cleanup;
    }
    printf ("+++[THREAD_SUB] [SUCCESS]: mosquitto_username_pw_set succeess\n\n");
	zlog_notice(zc, "+++[THREAD_SUB] mosquitto_username_pw_set succeess\n");
	
	/* 打印代理的host、port、keppalive */
	printf ("+++[THREAD_SUB] host: %s\n", mqtt->mosq_ctx.host);
	printf ("+++[THREAD_SUB] port: %d\n", mqtt->mosq_ctx.port);
	printf ("+++[THREAD_SUB] keepalive: %d\n\n", mqtt->mosq_ctx.keepalive);
	zlog_info(zc, "+++[THREAD_SUB] host: %s", mqtt->mosq_ctx.host);
	zlog_info(zc, "+++[THREAD_SUB] port: %d", mqtt->mosq_ctx.port);
	zlog_info(zc, "+++[THREAD_SUB] keepalive: %d\n", mqtt->mosq_ctx.keepalive);
	

    while ( 1 ) 
    {
        /* 客户端连接代理 */
        rv = mosquitto_connect(mosq, mqtt->mosq_ctx.host, mqtt->mosq_ctx.port, mqtt->mosq_ctx.keepalive);
        if (rv != MOSQ_ERR_SUCCESS)
        {
			/* 连接失败，3s后重连 */
            printf ("+++[THREAD_SUB] [ERROR]: connect failure, try reconnect after 3 seconds...\n");
			zlog_error(zc, "+++[THREAD_SUB] connect failure, try reconnect after 5 seconds...");
            sleep(3);
            printf ("+++[THREAD_SUB] 3 seconds, reconnect\n\n");
            zlog_notice(zc, "+++[THREAD_SUB] 3 seconds, reconnect\n");
			continue;
        }
		
		/*  */
        mosquitto_loop_forever(mosq, -1, 1);

        //printf ("+++[THREAD_SUB] [ERROR]: disconnect\n");
        //sleep(5);
        printf ("+++[THREAD_SUB] Reconnect\n\n");
		zlog_notice(zc, "+++[THREAD_SUB] Reconnect\n");
    
    }    
	
cleanup:
    mosquitto_destroy(mosq);
	printf ("+++[THREAD_SUB] Ending rv[%d]\n\n", rv);
	zlog_notice(zc, "+++[THREAD_SUB] Ending rv[%d]\n", rv);
	
	return NULL;
}

/* 订阅者连接回调函数 */
void sub_on_connect(struct mosquitto *mosq, void *obj, int rc)
{
    int                  mid;
    mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
    mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
	/* 打印连接状态，判断是否连接成功 */
    printf ("+++[THREAD_SUB] callback on_connect rc: %s\n", mosquitto_connack_string(rc));
	zlog_notice(zc, "+++[THREAD_SUB] callback on_connect rc: %s", mosquitto_connack_string(rc));
    if ( rc != 0 )
    {
		/* 连接失败 */
        printf ("+++[THREAD_SUB] [ERROR]: callback on_connect failure\n");
		zlog_error(zc, "+++[THREAD_SUB] callback on_connect failure\n");
        mosquitto_disconnect(mosq);
        return ;
    }
	/* 连接成功 */
    printf ("+++[THREAD_SUB] [SUCCESS]: mosquitto_connect success\n\n");
	zlog_notice(zc, "+++[THREAD_SUB] mosquitto_connect success\n");
	
	/* 打印要订阅的主题 */
	printf ("+++[THREAD_SUB] subscribe Topic: %s\n", mqtt->mosq_ctx.sub_topic);
	printf ("+++[THREAD_SUB] subscribe QoS: %d\n\n", mqtt->mosq_ctx.sub_qos);
	zlog_notice(zc, "+++[THREAD_SUB] Subscribed topic:%s Qos:%d", mqtt->mosq_ctx.sub_topic, mqtt->mosq_ctx.sub_qos);
	/* 发起订阅 */
    if ( mosquitto_subscribe(mosq, &mid, mqtt->mosq_ctx.sub_topic, mqtt->mosq_ctx.sub_qos) != MOSQ_ERR_SUCCESS ) 
    {
		/* 订阅失败 */
        printf ("+++[THREAD_SUB] [ERROR]: mosquitto_subscribe failure\n\n");
		zlog_error(zc, "+++[THREAD_SUB] mosquitto_subscribe failure\n");
        return ;
    }

}

/* 连接断开回调函数，目前只用作连接断开的提示 */
void sub_on_disconnect(struct mosquitto *mosq, void *obj, int rc)
{
	mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
    mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
    printf ("+++[THREAD_SUB] :( callback on_disconnect\n\n");
	zlog_notice(zc, "+++[THREAD_SUB] :( callback on_disconnect\n");
}

/* 订阅回调函数，目前只用作订阅成功的提示 */
void sub_on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{  
	mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
    mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
	printf ("+++[THREAD_SUB] [SUCCESS] callback on_subscribe\n");
	zlog_notice(zc, "+++[THREAD_SUB] Subscribed success, callback on_subscribe\n");
}

/* 消息回调函数，包含解析payload */
void sub_on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
	int                  rv = 0;
	mqtt_t              *mqtt= NULL;
	zlog_category_t     *zc = NULL;
	int                  rev_vlaue;
	
	mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
	/* 拷贝payload到mqtt->mosq_ctx.recv_payload中 */
	memset(mqtt->mosq_ctx.recv_payload, 0, sizeof(mqtt->mosq_ctx.recv_payload));
	strncpy(mqtt->mosq_ctx.recv_payload, (char *)message->payload, sizeof(mqtt->mosq_ctx.recv_payload));
	
	/* 打印接收的payload消息 */
	printf ("+++[THREAD_SUB] recv_payload: %s\n", mqtt->mosq_ctx.recv_payload);
	printf ("+++[THREAD_SUB] recv_payload_len: %d\n", strlen(mqtt->mosq_ctx.recv_payload));
	zlog_info(zc, "+++[THREAD_SUB] recv_payload: %s", mqtt->mosq_ctx.recv_payload);
	zlog_info(zc, "+++[THREAD_SUB] recv_payload_len: %d", strlen(mqtt->mosq_ctx.recv_payload));
	
	/* 判断消息是否为“quit” */
	if (strcmp(message->payload, "quit") == 0)
	{
		/* 断开与代理的连接 */
		mosquitto_disconnect(mosq);
		printf ("+++[THREAD_SUB] Received 'quit', Disconnect from the broker\n\n");
		zlog_notice(zc, "+++[THREAD_SUB] Received 'quit', Disconnect from the broker\n");
		return ;
	}
	
	/* 解析接收到的payload，看是不是针对 "LightSwitch" 的消息，并将值返回到rev_vlaue中 */
	rv = parse_payload(mqtt->mosq_ctx.recv_payload, mqtt->identifier.led, &rev_vlaue);
	if (rv < 0)
	{
		printf ("+++[THREAD_SUB] [Error]: parse_payload failure rv[%d]\n", rv);
		zlog_error(zc, "+++[THREAD_SUB] parse_payload failure rv[%d]\n", rv);
		return ;
	}
	printf("LightSwitch = %d\n", rev_vlaue);
	zlog_notice (zc, "LightSwitch = %d", rev_vlaue);
	
	/* 根据收到的值去执行不同的动作 */
	if (rev_vlaue == 1)
	{
		printf ("+++[THREAD_SUB] Open LED\n");
		zlog_notice(zc, "+++[THREAD_SUB] Open LED\n");
	}
	else if (rev_vlaue == 0)
	{
		printf ("+++[THREAD_SUB] Close LED\n");
		zlog_notice(zc, "+++[THREAD_SUB] Close LED\n");
	}
	else
	{
		printf ("+++[THREAD_SUB] value is not 0 or 1\n");
		zlog_notice(zc, "+++[THREAD_SUB] value is not 0 or 1\n");
	}
	
}

/* 发布子线程函数 */
void *pub_thread_func(void *thread_arg)
{
	int                  rv = 0;
	struct mosquitto    *mosq = NULL;
	mqtt_t              *mqtt = NULL;
	struct database_s   *DB = NULL;
	char                 sql[256];
	char                *errmsg = 0;
	zlog_category_t     *zc = NULL;

	/* 检查 thread_arg 参数是否为空 */
	if ( !thread_arg )
	{
		printf ("---[THREAD_PUB] [ERROR]: Invalid parameter in %s\n\n", __FUNCTION__);
		return NULL;
	}
	
	mqtt = (mqtt_t *)thread_arg;
	DB = &mqtt->DB;
	zc = mqtt->zlog.zc;

	/* 创建数据库 */
	rv = sqlite3_open(DB->db_name, &DB->db);
	if ( rv != 0 )
	{
		printf("---[THREAD_PUB] [ERROR]: Open database failure: %s\n", sqlite3_errmsg(DB->db));
		zlog_error(zc, "---[THREAD_PUB] Open database failure: %s", sqlite3_errmsg(DB->db));
		return NULL;
	}
	printf ("---[THREAD_PUB] [SUCCESS]: Open database success\n");
	zlog_notice(zc, "---[THREAD_PUB] Open database success");
	
	/* 创建表 */
	rv = sqlite_create_table(DB->db, DB->table_name);
	if ( rv < 0 )
	{
		printf ("---[THREAD_PUB] [ERROR]: Create Table failure: rv[%d]\n", rv);
		zlog_error(zc, "---[THREAD_PUB] Create Table failure: rv[%d]", rv);
		return NULL;
	}
	printf ("---[THREAD_PUB] [SUCCESS]: Create Table success\n");
	zlog_notice(zc, "---[THREAD_PUB] Create Table success");
	
	
	/* 新建mosquitto客户端，不保留消息，传入mqtt指针作为参数 */
	mosq = mosquitto_new(mqtt->mosq_ctx.clientid, true, (void *)mqtt);
    if (mosq == NULL)
    {
        printf ("---[THREAD_PUB] [ERROR]: mosquitto_new failure\n");
		zlog_error(zc, "---[THREAD_PUB] mosquitto_new failure");
        rv = -2;
        goto cleanup;
    }
    printf ("---[THREAD_PUB] [SUCCESS]: mosquitto_new success\n\n");
	zlog_notice(zc, "---[THREAD_PUB] mosquitto_new success");
    
    /* 设置回调函数（连接、断开、发布回调） */
	zlog_notice(zc, "+++[THREAD_PUB] set mosquitto callback\n");
    mosquitto_connect_callback_set(mosq, pub_on_connect);
    mosquitto_disconnect_callback_set(mosq, pub_on_disconnect);
    mosquitto_publish_callback_set(mosq, pub_on_publish); 

    /* 设置 username、password */
    printf ("---[THREAD_PUB] username: %s\n", mqtt->usr.name);
    printf ("---[THREAD_PUB] passwd: %s\n", mqtt->usr.passwd);
	zlog_info (zc, "---[THREAD_PUB] username: %s", mqtt->usr.name);
    zlog_info (zc, "---[THREAD_PUB] passwd: %s", mqtt->usr.passwd);
    rv = mosquitto_username_pw_set(mosq, mqtt->usr.name, mqtt->usr.passwd);
    if (rv != MOSQ_ERR_SUCCESS)
    {
        printf("---[THREAD_PUB] [ERROR]: mosquitto_username_pw_set failure\n\n");
		zlog_error (zc, "---[THREAD_PUB] mosquitto_username_pw_set failure\n");
        rv = -3;
        goto cleanup;
    }
    printf ("---[THREAD_PUB] [SUCCESS]: mosquitto_username_pw_set succeess\n\n");
    zlog_notice (zc, "---[THREAD_PUB] mosquitto_username_pw_set succeess\n");
	
	/* 打印代理的host、port、keppalive */
	printf ("---[THREAD_PUB] host: %s\n", mqtt->mosq_ctx.host);
	printf ("---[THREAD_PUB] port: %d\n", mqtt->mosq_ctx.port);
	printf ("---[THREAD_PUB] keepalive: %d\n\n", mqtt->mosq_ctx.keepalive);
	zlog_info (zc, "---[THREAD_PUB] host: %s", mqtt->mosq_ctx.host);
	zlog_info (zc, "---[THREAD_PUB] port: %d", mqtt->mosq_ctx.port);
	zlog_info (zc, "---[THREAD_PUB] keepalive: %d\n", mqtt->mosq_ctx.keepalive);

    while ( 1 )
    {
		/* 打包数据，丢进数据库 */
		/* 获取当前数据库中可用的最小id */
		rv = get_id(DB);
		if (rv < 0) 
		{
			printf ("---[THREAD_PUB] [ERROR]: get id failure: rv[%d]\n", rv);
			zlog_error (zc, "---[THREAD_PUB] get id failure: rv[%d]", rv);
		}
		printf ("---[THREAD_PUB] [SUCCESS]: get id success id[%d]\n", DB->id);
		zlog_info (zc, "---[THREAD_PUB] get id success id[%d]", DB->id);
		
		/* 获取DS18B20温度 */
		rv = Get_DS18B20_Temp(&DB->temp);
		if (rv < 0)
		{
			printf ("---[THREAD_PUB] [ERROR]: get ds18b20 Temperature failure, rv[%d]\n", rv);
			zlog_error (zc, "---[THREAD_PUB] get ds18b20 Temperature failure, rv[%d]", rv);
			continue;
			//return NULL;
		}
		printf("---[THREAD_PUB] [SUCCESS]: get ds18b20 Temperature: %.2f C\n", DB->temp);
		zlog_info (zc, "---[THREAD_PUB] get ds18b20 Temperature: %.2f C", DB->temp);
		
		/* 获取当前时间 */
		memset (DB->timestamp, 0, sizeof(DB->timestamp));
		strncpy (DB->timestamp, get_cur_time(), sizeof(DB->timestamp));
		printf ("---[THREAD_PUB] [SUCCESS]: get timestamp: %s\n\n", DB->timestamp);
		zlog_info (zc, "---[THREAD_PUB] get timestamp: %s\n", DB->timestamp);
		
		/* 
		 * 将数据插入数据库的表中
		 * 若表满了，则按顺序更新数据
		 * 若表未满，则直接插入数据
		 */
		memset (sql, 0, sizeof(sql));
		if ( DB->full_flag )	//表满了，要更新数据
		{
			printf ("---[THREAD_PUB] Table %s is full, update data\n", DB->table_name);
			zlog_info (zc, "---[THREAD_PUB] Table %s is full, update data", DB->table_name);
			rv = update_data (DB->db, DB->table_name, &DB->id, &DB->temp, DB->timestamp);
			if (rv < 0)
			{
				//更新失败，退出线程
				printf ("---[THREAD_PUB] [ERROR]: update data failure: rv[%d]\n", rv);
				zlog_error (zc, "---[THREAD_PUB] update data failure: rv[%d]\n", rv);
				//continue;
				//return -1;
				rv = -5;
				goto cleanup;
			}
			printf ("---[THREAD_PUB] [SUCCESS]: update data success\n\n");
			zlog_notice (zc, "---[THREAD_PUB] update data success\n");
			
		}
		else  					//表没满，直接插入
		{
			printf ("---[THREAD_PUB] Table %s is not full, insert data\n", DB->table_name);
			zlog_info (zc, "---[THREAD_PUB] Table %s is not full, insert data", DB->table_name);
            //插入数据
			rv = insert_data (DB->db, DB->table_name, &DB->id, &DB->temp, DB->timestamp);
			if (rv < 0)
			{
				//数据插入失败，退出线程
				printf ("---[THREAD_PUB] [ERROR]: insert data failure: rv[%d]\n\n", rv);
				zlog_error (zc, "---[THREAD_PUB] insert data failure: rv[%d]\n", rv);
				
				//return -1;
				rv = -6;
				goto cleanup;
			}
			printf ("---[THREAD_PUB] [SUCCESS]: insert data success\n\n");
			zlog_notice (zc, "---[THREAD_PUB] insert data success\n");
			
		}	
		
        /* 连接代理 */
        rv = mosquitto_connect(mosq, mqtt->mosq_ctx.host, mqtt->mosq_ctx.port, mqtt->mosq_ctx.keepalive);
        if (rv != MOSQ_ERR_SUCCESS)
        {
			/* 连接失败，5s后重连 */
            printf ("---[THREAD_PUB] [ERROR]: Connect failure, try Reconnect after 5 second\n");
            zlog_error (zc, "---[THREAD_PUB] Connect failure, try Reconnect after 5 second");
            sleep(5);
            printf ("---[THREAD_PUB] 5 seconds,  Reconnect\n\n");
            zlog_notice (zc, "---[THREAD_PUB] 5 seconds,  Reconnect\n");
            continue;
        }
		
        mosquitto_loop_forever(mosq, -1, 1);
		
		/* 连接断开，5s后重连 */
        printf ("---[THREAD_PUB] Sleep 5 seconds...\n");
		zlog_notice (zc, "---[THREAD_PUB] Sleep 5 seconds...\n");
        sleep(5);
        printf ("---[THREAD_PUB] 5 Seconds Reconnect\n\n");
        zlog_notice (zc, "---[THREAD_PUB] 5 Seconds Reconnect\n");
    
    }    
	
cleanup:
    mosquitto_destroy(mosq);
	sqlite3_close(DB->db);
	printf ("---[THREAD_PUB] Ending rv[%d]\n\n", rv);
	zlog_notice (zc, "---[THREAD_PUB] Ending rv[%d]\n", rv);
	
	return NULL;
}

/* 连接回调函数，采用短连接，连接成功后，发布消息，随即断开连接 */
void pub_on_connect(struct mosquitto *mosq, void *obj, int rc)
{
	int                  rv = 0;
    int                  mid;
    mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
    mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
	/* 打印连接状态，判断是否连接成功 */
    printf ("---[THREAD_PUB] callback on_connect rc: %s\n", mosquitto_connack_string(rc));
    zlog_info (zc, "---[THREAD_PUB] callback on_connect rc: %s", mosquitto_connack_string(rc));
    if ( rc != 0 )
    {
		/* 连接失败，打印错误信息，客户端断开连接，立即返回 */
        printf ("---[THREAD_PUB] [ERROR]: callback on_connect failure\n");
        zlog_error (zc, "---[THREAD_PUB] connect failure");
        mosquitto_disconnect(mosq);
        return ;
    }
    printf ("---[THREAD_PUB] [SUCCESS]: mosquitto_connect success\n");
    zlog_notice (zc, "---[THREAD_PUB] mosquitto_connect success");
	
	/* 发布消息 */
	rv = upload(mosq, mqtt);
	if ( rv < 0 )
	{
		printf ("---[THREAD_PUB] [ERROR]: Publish failure\n\n");
		zlog_error (zc, "---[THREAD_PUB] Publish failure\n");
	}
	
	/* 断开连接 */
	mosquitto_disconnect(mosq);
	
}

/* 连接断开回调函数，目前只用作断开连接的提示 */
void pub_on_disconnect(struct mosquitto *mosq, void *obj, int rc)
{
	mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
    mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
    printf ("---[THREAD_PUB] callback on_disconnect\n\n");
    zlog_notice (zc, "---[THREAD_PUB] callback on_disconnect\n");
}

/* 发布回调函数，说明消息已发布成功，从数据库中删除数据 */
void pub_on_publish(struct mosquitto *mosq, void *obj, int mid)
{
	int                  rv = 0;
	mqtt_t              *mqtt = NULL;
	zlog_category_t     *zc = NULL;
	
	mqtt = (mqtt_t *)obj;
	zc = mqtt->zlog.zc;
	
    printf ("---[THREAD_PUB] [SUCCESS]: callback on_publish\n");
	printf ("---[THREAD_PUB] message %d has been posted to the cloud\n", mid);
	//printf ("---[THREAD_PUB] pub_payload:\n%s\n\n", mqtt->mosq_ctx.pub_payload);
	zlog_info (zc, "---[THREAD_PUB] callback on_publish");
	zlog_info (zc, "---[THREAD_PUB] message %d has been posted to the cloud", mid);
	//zlog_info (zc, "---[THREAD_PUB] pub_payload:\n%s\n", mqtt->mosq_ctx.pub_payload);
	
	/* 从数据表中删除数据 */
	printf ("---[THREAD_PUB] Delete the corresponding data from the table[%s]\n", mqtt->DB.table_name);
	zlog_notice (zc, "---[THREAD_PUB] Delete the corresponding data from the table[%s]", mqtt->DB.table_name);
	rv = delete_data (mqtt->DB.db, mqtt->DB.table_name, &mqtt->DB.id);
	if (rv < 0)
	{
		printf ("---[THREAD_PUB] [ERROR] delete data failure: rv[%d]\n\n", rv);
		zlog_error (zc, "---[THREAD_PUB] delete data failure: rv[%d]\n", rv);
		return ;
	}
	printf ("---[THREAD_PUB] [SUCCESS] delete data success\n\n");
	zlog_notice (zc, "---[THREAD_PUB] delete data success\n");

	return ;
}

/* 上报温度信息的函数 */
int upload (struct mosquitto *mosq, struct mqtt_s *mqtt)
{
	int                  rv = 0;
	int                  mid;
	char                *msg = NULL;
	struct mosq_ctx_s   *mosq_ctx = NULL;
	struct database_s   *DB = NULL;
	zlog_category_t     *zc = NULL;
	
	if ( !mosq || !mqtt )
	{
		printf ("---[THREAD_PUB] [ERROR]: Invalid parameter in %s\n\n", __FUNCTION__);
		return -1;
	}
	
	mosq_ctx = &mqtt->mosq_ctx;
	DB = &mqtt->DB;
	zc = mqtt->zlog.zc;
	

	/* 打包成JSON格式数据 */
	memset(mosq_ctx->pub_payload, 0, strlen(mosq_ctx->pub_payload));
	msg = pack_cjon_float(&mqtt->payload_format, mqtt->identifier.temp, &DB->temp);
	if ( msg == NULL )
	{
		printf ("---[THREAD_PUB] [ERROR]: pack_cjon_float failure, rv[%d]\n", rv);
		zlog_error (zc, "---[THREAD_PUB] pack_cjon_float failure, rv[%d]", rv);
		return -3;
	}
	strncpy(mosq_ctx->pub_payload, msg, sizeof(mosq_ctx->pub_payload));
	printf ("pub_payload:\n");
	printf ("%s\n\n", msg);
	
	/* 发布 */
	rv = mosquitto_publish(mosq, &mid, mosq_ctx->pub_topic, strlen(msg), msg, mosq_ctx->pub_qos, false);
	if ( rv != MOSQ_ERR_SUCCESS )
	{
		printf("---[THREAD_PUB] [ERROR]: mosquitto_publish failure: %s\n", strerror(errno));
		zlog_error(zc, "---[THREAD_PUB] mosquitto_publish failure: %s\n", strerror(errno));
		return -4;
	}
	
	return 0;
}
